package com.Reactor3参考文档;

import java.util.concurrent.atomic.AtomicLong;

import org.junit.Test;

import reactor.core.publisher.Flux;

/**最简单的创建 Flux 的方式就是使用 generate 方法。

这是一种 同步地， 逐个地 产生值的方法，意味着 sink 是一个 SynchronousSink 而且其 next() 方法在每次回调的时候最多只能被调用一次。你也可以调用 error(Throwable) 或者 complete()，不过是可选的。

最有用的一种方式就是同时能够记录一个状态值（state），从而在使用 sink 发出下一个元素的时候能够 基于这个状态值去产生元素。此时生成器（generator）方法就是一个 BiFunction<S, SynchronousSink<T>, S>， 其中 <S> 是状态对象的类型。你需要提供一个 Supplier<S> 来初始化状态值，而生成器需要 在每一“回合”生成元素后返回新的状态值（供下一回合使用）。

例如我们使用一个 int 作为状态值。
 * @author DELL
 *
 */
public class Generate {

	@Test
	public void test() {
		// 初始化状态值（state）为0。
		Flux<String> flux = Flux.generate(() -> 0, (state, sink) -> {
			// 我们基于状态值 state 来生成下一个值（state 乘以 3）。
			sink.next("3 x " + state + " = " + 3 * state);
			// 我们也可以用状态值来决定什么时候终止序列。
			if (state == 10)
				sink.complete();
			// 返回一个新的状态值 state，用于下一次调用。
			return state + 1;
		});
		flux.subscribe(System.out::println);
	}

	@Test
	// 我们也可以使用可变（mutable）类型（译者注：如上例，原生类型及其包装类，以及String等属于不可变类型） 的 <S>。上边的例子也可以用 AtomicLong 作为状态值，在每次生成后改变它的值。
	public void test2() {
		Flux<String> flux = Flux.generate(AtomicLong::new, (state, sink) -> {
			long i = state.getAndIncrement();
			sink.next("3 x " + i + " = " + 3 * i);
			if (i == 10)
				sink.complete();
			return state;
		});
	}

	@Test
	// 如果状态对象需要清理资源，可以使用 generate(Supplier<S>, BiFunction, Consumer<S>) 这个签名方法来清理状态对象（译者注：Comsumer 在序列终止才被调用）。
	// 如果 state 使用了数据库连接或者其他需要最终进行清理的资源，这个 Consumer lambda 可以用来在最后关闭连接或完成相关的其他清理任务。
	public void test3() {
		Flux<String> flux = Flux.generate(AtomicLong::new, (state, sink) -> {
			long i = state.getAndIncrement();
			sink.next("3 x " + i + " = " + 3 * i);
			if (i == 10)
				sink.complete();
			return state;
			// 我们会看到最后一个状态值（11）会被这个 Consumer lambda 输出。
		}, (state) -> System.out.println("state: " + state));
		flux.subscribe();
	}

}
